package org.sample.reactive;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.stream.Stream;

public class ReactiveSample {

	private static final Logger LOGGER = LoggerFactory.getLogger(ReactiveSample.class);

	public static void main(String[] args) {
		logAndConvert();
	}

	public static void logAndConvert() {
		Flux<String> flux = Flux.just("red", "white", "blue").log();
		Flux<String> upperFlux = flux.map(String::toUpperCase);
		upperFlux.subscribe(System.out::println);

		Stream<String> stream = Stream.of("red", "white", "blue");
		Stream<String> upperStream = stream.map(value -> {
			System.out.println(value);
			return value.toUpperCase();
		});
		upperStream.forEach(System.out::println);
	}

	public static void bySubscriber() {
		Flux<String> flux = Flux.just("red", "white", "blue").log();
		Flux<String> upperFlux = flux.map(String::toUpperCase);

		upperFlux.subscribe(new Subscriber<String>() {

			private long count = 0;
			private Subscription subscription;

			@Override
			public void onSubscribe(Subscription s) {
				this.subscription = s;
				subscription.request(2);
			}

			@Override
			public void onNext(String s) {
				count++;
				if (count >= 2) {
					count = 0;
					subscription.request(2);
				}
			}

			@Override
			public void onError(Throwable t) {

			}

			@Override
			public void onComplete() {

			}
		});
	}

	public static void requestBound() {
		Flux<String> flux = Flux.just("red", "white", "blue").log();
		Flux<String> upperFlux = flux.map(String::toUpperCase);
		upperFlux.every(2).subscribe(System.out::println);
	}

	public static void subscribeParallel() {
		Flux<String> flux = Flux.just("red", "white", "blue").log();
		Flux<String> upperFlux = flux.map(String::toUpperCase);
		upperFlux.every(2).subscribeOn(Schedulers.parallel()).subscribe(LOGGER::info);

		try {
			Thread.sleep(3000);
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public static void subscribeMultipleThread() {
		Flux<String> flux = Flux.just("red", "white", "blue", "yellow", "green", "violet", "purple", "orange", "cyan")
				.log();
		Flux<String> upperFlux = flux
				.flatMap(value -> Mono.just(value.toUpperCase()).subscribeOn(Schedulers.parallel()), 2);
		upperFlux.subscribe(LOGGER::info);

		try {
			Thread.sleep(3000);
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public static void publishMultipleThread() {
		Flux<String> flux = Flux.just("red", "white", "blue", "yellow", "green", "violet", "purple", "orange", "cyan")
				.log();
		Flux<String> upperFlux = flux.map(String::toUpperCase).subscribeOn(Schedulers.newParallel("sub"))
				.publishOn(Schedulers.newParallel("pub"));
		upperFlux.subscribe(LOGGER::info);
	}

}
